package com.run.fjy.mr;

import javax.xml.soap.Text;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;

import com.run.bcpimp.mr.BcpExtractMR;
import com.run.bcpimp.mr.io.ExtractTableValueWritable;
import com.run.bcpimp.util.RunTool;
import com.run.fjy.io.TerminalWritable;


public class NewAddLocal extends RunTool
{

	public NewAddLocal()
	{
	}

	public int exec(String arg0[])
		throws Exception
	{
		Configuration conf = getConf();
		conf.set("cfgfile.bcpextract", "conf/bcpextract_newadd.xml");
		Job job = new Job(conf);
		job.setJarByClass(NewAddLocal.class);
		job.setMapperClass(BcpExtractMR.MyMapperBase.class);
		job.setCombinerClass(NewAddLocalCombiner.class);
		job.setReducerClass(NewAddLocalReducer.class);
		job.setMapOutputValueClass(ExtractTableValueWritable.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(TerminalWritable.class);
		job.setInputFormatClass(SequenceFileInputFormat.class);
		setInOutPaths(job, arg0);
		job.setOutputFormatClass(SequenceFileOutputFormat.class);
		System.exit(job.waitForCompletion(true) ? 0 : 1);
		return 0;
	}

	protected int getArgumentNumber()
	{
		return 2;
	}
}